package com.k8.registry;


import com.k8.common.event.EventDeployer;
import com.k8.common.executor.ExecutorRepository;
import com.k8.rpc.registry.Registry;
import com.k8.rpc.RpcException;
import com.k8.rpc.exchange.Exchange;
import com.k8.rpc.exchange.ExchangeManager;
import com.k8.rpc.exchange.impl.DefaultExchangeManager;
import com.k8.rpc.expoter.AbstractExporter;
import com.k8.rpc.expoter.Exporter;
import com.k8.rpc.invoke.Invoker;
import com.k8.rpc.invoke.InvokerManager;
import com.k8.rpc.url.ApiSignature;
import com.k8.rpc.url.URL;
import com.k8.rpc.util.URLUtils;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.Lock;

import static com.k8.common.constants.CommonKeyConstants.RegistryKeyConstant.REGISTRY_KEY;

@Slf4j
public class K8Protocol extends AbstractRegistryProtocol {
    private volatile boolean isDestroy;
    @Resource
    private InvokerManager invokerManager;
    @Resource
    private EventDeployer eventDeployer;
    private volatile boolean isFirstOpen;
    private ExchangeManager exchangeManager;
    /**
     * 默认使用 {@link com.k8.common.executor.DefaultExecutorRepository}
     */
    @Resource
    private ExecutorRepository executorRepository;


    public void setExecutorRepository(ExecutorRepository executorRepository) {
        this.executorRepository = executorRepository;
    }

    public void setInvokerManager(InvokerManager invokerManager) {
        this.invokerManager = invokerManager;
    }

    public void setEventDeployer(EventDeployer eventDeployer) {
        this.eventDeployer = eventDeployer;
    }

    /**
     * 该协议需要在服务导出做的事，区分不同协议的点
     */
    @Override
    protected <T> Exporter<T> doExport(Invoker<T> invoker, URL registerUrl, URL protocolUrl, Registry registry, ApiSignature apiSignature) throws RpcException {
        Exporter<T> exporter = new DefaultExporter<>(executorRepository.getExecutorService(protocolUrl), invoker, registry, registerUrl, eventDeployer, apiSignature);
        try {
            openServer(protocolUrl);
        } catch (ExecutionException | InterruptedException e) {
            throw new RpcException(e);
        }
        return exporter;
    }

    @Override
    public void destroy() {
        if (isDestroy) return;
        synchronized (this) {
            if (isDestroy) return;
            exchangeManager.shutDown();
            isDestroy = true;
        }
    }

    private void openServer(URL protocolUrl) throws ExecutionException, InterruptedException {
        firstOpen();
        Exchange exchange = exchangeManager.getOrRegistryExchange(URLUtils.getFullPath(protocolUrl), protocolUrl);
        if (!exchange.serverOpened()) {
            exchange.openServer(protocolUrl, beanFactory);
        }
    }

    private void firstOpen() {
        if (isFirstOpen) return;
        synchronized (this) {
            if (isFirstOpen) return;
            exchangeManager = beanFactory.getOrRegistryBean(DefaultExchangeManager.class);
            isFirstOpen = true;
        }
    }


    /**
     * 协议需要实现自己的消费者客户端启动，super{@link AbstractRegistryProtocol#refer(URL, Class)}
     * 支持根据url以及协议生成对应的apiSignature从注册中心获取相应配置转化为protocolUrl，指导每次调用时连接的创建
     * 提供者端一个应用的某个服务可能支持多个协议，但是应用不管有多少个实例，某一个协议的配置应该是唯一的，不会随不同的服务而变化
     * 而消费者端refer时只需要加载本服务所需协议（消费者端只能配置一个协议）的protocol配置，再检测应用的简单元数据中是否支持该协议
     */
    @Override
    public <T> Invoker<T> refer(URL url, Class<T> serviceType) throws RpcException {
        //先进行协议头的切换
        URL registerConfigUrl = (URL) url.getAttribute(REGISTRY_KEY);
        String invokerKey = registerConfigUrl.getSimpleString();
        Invoker invoker = this.invokerManager.getInvoker(invokerKey);
        if (invoker == null) {
            Lock lock = this.invokerManager.getLock();
            try {
                lock.lock();
                invoker = this.invokerManager.getInvoker(invokerKey);
                if (invoker == null) {
                    //todo 创建invoker,为支持一个注册中心有多个实例，故此依然需要一个directory来维护多个实例
                    //故此直接从父类支持注册中心获取配置的protocol来具体生成invoker，但是又需要暴露接口对每个实例开启客户端
                    invoker = super.refer(url, serviceType);
                    invokerManager.putInvoker(invokerKey, invoker);
                }
            } finally {
                lock.unlock();
            }
        }
        return invoker;
    }


    public static class DefaultExporter<T> extends AbstractExporter<T> {

        private Invoker invoker;
        private Registry registry;

        private URL registerUrl;

        public DefaultExporter(ExecutorService executorService, Invoker<T> invoker, Registry registry, URL registerUrl, EventDeployer eventDeployer, ApiSignature apiSignature) {
            super(executorService, eventDeployer, apiSignature);
            assert invoker != null;
            assert registry != null;
            assert registerUrl != null;
            this.invoker = invoker;
            this.registry = registry;
            this.registerUrl = registerUrl;
        }

        @Override
        public Invoker<T> getInvoker() {
            return this.invoker;
        }

        @Override
        public void unexport() {

        }

        @Override
        public void doRegister() {
            //todo 这里需要增强，整合metrics,同时需要拆分服务注册和配置注,并且导出的类型这里也需要变
            registry.registerMappingAndConfiguration(registerUrl);
        }

        @Override
        public void unregister() {
            //todo
        }
    }
}
